package com.magicalcoder.pigeon.zookeeper;

import com.magicalcoder.pigeon.common.callback.ZookeeperNodeWatchCallback;
import com.magicalcoder.pigeon.common.dto.RmtAddress;
import com.magicalcoder.pigeon.common.util.RmtAddressUtil;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * Created by hzhedongyu on 2017/5/31.
 * 客户端寻找可用服务节点
 */
public class ClientFoundService {

    private Logger logger = LoggerFactory.getLogger(ClientFoundService.class);

    private ClientFoundService(){

    }

    private static final ClientFoundService instance = new ClientFoundService();
    public static ClientFoundService get(){
        return instance;
    }

    private volatile ZooKeeper zooKeeper = null;

    //服务器列表 host:port必须唯一
    private volatile CopyOnWriteArrayList<RmtAddress> serviceList = new CopyOnWriteArrayList<RmtAddress>();

    public void watchZookeeper(ZookeeperNodeWatchCallback callback){
        connect(callback);
    }

    //监控节点编号 负载 debug会出现问题 run测试便可
    private void watchZkNodeChange(ZookeeperNodeWatchCallback callback){
        try {
            List<String> children = zooKeeper.getChildren(ZooConstant.ROOT, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    logger.debug("client 监听到root节点变更");
                    if (event.getType() == Event.EventType.NodeChildrenChanged) {//根节点发生变化 当有节点发生变动 这里就执行了
                        watchZkNodeChange(callback);
                    }
                }
            });
            if(children!=null && children.size()>0){
                CopyOnWriteArrayList<RmtAddress> newList = new CopyOnWriteArrayList<RmtAddress>();
                for(String path:children){
                    byte[] data = zooKeeper.getData(ZooConstant.ROOT+"/"+path,false,null);
                    newList.add(RmtAddressUtil.cover(new String(data)));
                }
                serviceList = newList;
            }else {
                serviceList = new CopyOnWriteArrayList<RmtAddress>();
            }
            callback.call(serviceList);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    public CopyOnWriteArrayList<RmtAddress> getServiceList() {
        return serviceList;
    }

    public void setServiceList(CopyOnWriteArrayList<RmtAddress> serviceList) {
        this.serviceList = serviceList;
    }

    private void connect(ZookeeperNodeWatchCallback callback) {
        final CountDownLatch latch = new CountDownLatch(1);
        try {
            String server = ZookeeperProperties.getServer();
            zooKeeper = new ZooKeeper(server, 5000, new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {
                    if(watchedEvent.getState() == Event.KeeperState.SyncConnected){
                        latch.countDown();
                    }else if(watchedEvent.getState() == Event.KeeperState.Expired){
                        logger.error("client SESSION失效，失去连接,尝试重新连接");
                        reconnect(callback);
                    }else if(watchedEvent.getState() == Event.KeeperState.Disconnected){
                        logger.error("client 失去连接,尝试重新连接");
                        reconnect(callback);
                    }else {
                        logger.info("client zookeeper"+watchedEvent.getState());
                    }
                }
            });
        } catch (IOException e) {
            e.printStackTrace();
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        watchZkNodeChange(callback);
        logger.info("client 连接成功");
    }

    private void reconnect(ZookeeperNodeWatchCallback callback){
        try {
            zooKeeper.close();
            TimeUnit.SECONDS.sleep(1);
            connect(callback);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}
